package com.lm.flink.log;

import com.lm.flink.entry.LogLevelCount;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class RedisLevelCountSink extends RichSinkFunction<LogLevelCount> {
    private transient JedisPool jedisPool;

    @Override
    public void open(Configuration parameters) throws Exception {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(10);
        poolConfig.setMaxIdle(5);
        poolConfig.setMinIdle(1);

        if(jedisPool == null){
            jedisPool = new JedisPool(poolConfig, "10.2.0.230", 6379, 2000, "123456");
        }
    }

    @Override
    public void invoke(LogLevelCount count, Context context) throws Exception {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.select(0);

            // 存储实时计数
            String key = "log:level:" + count.getLevel() + ":count";
            jedis.set(key, String.valueOf(count.getCount()));

            // 存储时间序列数据
            String tsKey = "log:level:" + count.getLevel() + ":timeline";
            jedis.zadd(tsKey, count.getWindowEnd(), String.valueOf(count.getCount()));

            // 更新总览
            jedis.hset("log:stats", "level:" + count.getLevel(), String.valueOf(count.getCount()));

            System.out.println("Updated level count: " + count.getLevel() + " = " + count.getCount());
        }
    }

    @Override
    public void close() throws Exception {
        if (jedisPool != null) {
            jedisPool.close();
        }
    }
}
